package com.linjing.demo.message.mq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;


@Component
public class MQConsumer {

    private DefaultMQPushConsumer consumer;

    private String nameAddr = "192.168.146.138:9876;192.168.146.134:9876;192.168.146.128:9876;192.168.146.135:9876";

    private String topicName = "myTopic";


    @PostConstruct
    public void init() throws MQClientException {
        consumer = new DefaultMQPushConsumer("MyGroup");
        consumer.setNamesrvAddr(nameAddr);
        consumer.subscribe(topicName, "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                //实现库存真正到数据库内扣减的逻辑
                for (Message msg : msgs) {
                    String jsonString = new String(msg.getBody());
                    System.out.println("接受减库存消息: " + jsonString);
                    // 幂等消费
                    System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
                    System.out.println("业务处理");
                    System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

            }
        });

        consumer.start();

    }
}
